%%%====================================================================
%%% File    : pipe_sender.erl
%%% Author  :  <vjache>
%%% Description: This is a simpliest pipe implementation. It may be 
%%%				 considered as a simple reference implementation of 
%%%				 a pipe callback module, and could be used for debug 
%%%				 purposes.
%%%
%%% Created : 24.03.2009
%%%====================================================================

-module(pipe_sender).

%%
%% Include files
%%
-include("../include/demetra.hrl").
-include("../include/pipe.hrl").
-include("../include/pgt.hrl").
%%
%% Exported Functions
%%
-export([
		 new/2,
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 test/1]).

-define(THRESHOLD,100).



%%====================================================================
%% API Functions
%%====================================================================

new() ->
	new(fun(_Node)->
				self()
		end,
		make_ref()).

new(Node2PrcFun,Signature) ->
	pipe:pipe_from_module(?MODULE, 
						  _Param={Node2PrcFun,Signature}, 
						  _NextPipe=terminal).

open({{Node2PrcFun,Signature},terminal=_NextPipe}) -> %%PipeState
    {Node2PrcFun,Signature,0}.
push(S,?EOF=_Item) ->
	S;
push({Node2PrcFun,Signature,MsgSentCount},Item) 
  when MsgSentCount>=?THRESHOLD ->
	push({Node2PrcFun,Signature,cfm_clean(MsgSentCount,0)},Item);
push({Node2PrcFun,Signature,MsgSentCount},{Node,Value}=_Item) ->
	Node2PrcFun(Node)!?PIPE_SENDER_MSG(Signature,self(),Value),
%% 	io:format("DBG SENDER(node=~p,pid=~p,to=~p) ~p!~p",[node(),self(),?PARENT_PG,Node2PrcFun(Node),?PIPE_SENDER_MSG(Signature,self(),Value)]),
	{Node2PrcFun,Signature,MsgSentCount+1}.
flush(S) ->
	S.
close({_Node2PrcFun,_Signature,MsgSentCount}) ->
	case cfm_clean(MsgSentCount,?GENERAL_TIMEOUT) of
		0 ->
			ok;
		LostCount ->
			exit({confirm_msg_timeout,LostCount,?GENERAL_TIMEOUT})
	end.

%%====================================================================
%% Test Functions
%%====================================================================

test(L) ->
	S=pipe:open(new()),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).


%%====================================================================
%% Local Functions
%%====================================================================
cfm_clean(0,_Timeout) ->
	0;
cfm_clean(Count,Timeout) ->
	receive 
		?PIPE_SENDER_MSG_CONFIRM ->
		  cfm_clean(Count-1,Timeout)
	after Timeout ->
		Count
	end.
	
